package com.emqx.kafka.demo.service.impl;

import com.emqx.kafka.demo.service.IKafkaService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;


/**
 * kafka消息服务发送
 *
 * @author David
 * @date 2022/4/12 10:12
 */
@Service
public class KafkaServiceImpl implements IKafkaService {

    private static final Logger log = LoggerFactory.getLogger(KafkaServiceImpl.class);

    @Autowired
    private StreamBridge bridge;

    private final static String BRACE_PREFIX = "{";

    private final static String BRACE_SUFFIX = "}";

    private final static String BRACKET_PREFIX = "[";

    private final static String BRACKET_SUFFIX = "]";

    @Override
    public void sendMessageByConfig(String mqTopic, String messageJson) {
        if (!StringUtils.hasText(mqTopic)) {
            log.info("根据配置发送kafka消息,配置mqTopic参数为空,无需发送!");
            return;
        }

        if (!isJSONType(messageJson)) {
            log.error("根据配置发送kafka消息,根据配置名称{},发送体类型不是json无需发送{}!", mqTopic, messageJson);
            return;
        }

        //mqtopic名称与kafkatopic名称保持一致   不能用/隔开命名  kafkatopic /隔开会报错
        try {
            bridge.send(mqTopic.replace('/','-'), messageJson);
        } catch (Exception e) {
            log.error("发送kafka通道报错{},请检查配置!{}", mqTopic, e);
        }
    }

    private static boolean isJSONType(String str) {
        if (StringUtils.isEmpty(str)) {
            return false;
        }
        str = str.trim();
        if (str.startsWith(BRACE_PREFIX) && str.endsWith(BRACE_SUFFIX)) {
            return true;
        } else if (str.startsWith(BRACKET_PREFIX) && str.endsWith(BRACKET_SUFFIX)) {
            return true;
        }
        return false;
    }
}
